package com.lizhiyu.flink.demo2_source.paramlleism;

import com.lizhiyu.flink.model.VideoOrder;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 通过不同的并行度来处理数据
 * 如下程序 source 的并行度为2    filter的并行度为3   sink的并行度为 4 则最大并行度取最大的
 * localhost:8081访问
 */
public class ParallelismSource {

    public static void main(String[] args) throws Exception {
        //构建执行任务环境以及任务的启动的入口, 存储全局相关的参数
        //StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        env.setParallelism(2);
        DataStream<VideoOrder> videoOrderDS =  env.addSource(new VideoCustomSource());

        DataStream<VideoOrder> filterDS = videoOrderDS.filter(new FilterFunction<VideoOrder>() {
            @Override
            public boolean filter(VideoOrder videoOrder) throws Exception {
                return videoOrder.getMoney()>5;
            }
        }).setParallelism(3);

        filterDS.print().setParallelism(4);
        //DataStream需要调用execute,可以取个名称
        env.execute("source job");
    }
}
